package com.gandong8.hbase.tool.util;


import com.gandong8.hbase.tool.annotation.HbaseColumn;
import com.gandong8.hbase.tool.domain.ConstantDO;
import com.gandong8.hbase.tool.domain.HbaseCell;
import com.gandong8.hbase.tool.domain.HbaseFieldTypeEnum;
import com.gandong8.hbase.tool.exception.TypeNotSupportException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.*;

@Slf4j
public class HbaseUtil {

    /**
     * map 转 put
     * @param map
     * @return
     */
    public static Put mapToPut(Map<String, String> map) {
        Put put = new Put(Bytes.toBytes(map.get(ConstantDO.ROW_KEY)));
        map.forEach((key, value) -> {
            String[] arr = key.split(":");
            put.addColumn(Bytes.toBytes(arr[0]), Bytes.toBytes(arr[1]), Bytes.toBytes(value));
        });
        return put;
    }

    /**
     * object 转 put
     * @param obj
     * @param <T>
     * @return
     * @throws Exception
     */
    public static <T> Put objectToPut(T obj) throws Exception {
        return objectToPut(obj, ConstantDO.DEFAULT_ROW);
    }

    /**
     * object 转 put
     * @param obj
     * @param rowKey
     * @param <T>
     * @return
     * @throws Exception
     */
    public static <T> Put objectToPut(T obj, String rowKey) throws Exception {
        Put put = new Put(Bytes.toBytes(Objects.requireNonNull(getRowKey(obj, rowKey))));
        Field[] fields = obj.getClass().getDeclaredFields();
        for (Field field : fields) {
            HbaseColumn column = field.getAnnotation(HbaseColumn.class);
            if(column == null) {
                continue;
            }
            String typeName = field.getType().getTypeName();
            if(!HbaseFieldTypeEnum.contains(typeName)) {
                throw new TypeNotSupportException(field.getName());
            }
            field.setAccessible(true);
            String family = column.family();
            String qualifier = column.qualifier();
            if (ConstantDO.ROW_KEY.equalsIgnoreCase(family)) {
                continue;
            }
            if (StringUtils.isBlank(family) || StringUtils.isBlank(qualifier)) {
                continue;
            }
            if(HbaseFieldTypeEnum.DATE.getTypeName().equals(typeName)) {
                Date fieldObject = (Date) field.get(obj);
                if(fieldObject == null) {
                    continue;
                }
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(String.valueOf(fieldObject.getTime())));
                continue;
            }
            Object fieldObject = field.get(obj);
            if(fieldObject == null) {
                continue;
            }
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(fieldObject.toString()));
        }
        return put;
    }

    /**
     * 获取对象的rowkey值
     * @param obj
     * @param <T>
     * @return
     */
    public static <T> String getRowKey(T obj) {
        return getRowKey(obj, ConstantDO.DEFAULT_ROW);
    }

    /**
     * 获取对象的rowkey值
     * @param obj
     * @param rowKey
     * @param <T>
     * @return
     */
    private static <T> String getRowKey(T obj, String rowKey) {
        Class<?> clazz = obj.getClass();
        try {
            Field field = clazz.getDeclaredField(rowKey);
            field.setAccessible(true);
            Object object = field.get(obj);
            return object.toString();
        } catch (NoSuchFieldException e) {
            log.error("NoSuchFieldException:{}", rowKey);
        } catch (Exception e) {
            log.error("getRowObject_err", e);
        }
        return null;
    }

    /**
     * result转object
     * @param result
     * @param clazz
     * @param <T>
     * @return
     * @throws Exception
     */
    public static <T> T resultToObject(Result result, Class<T> clazz) throws Exception {
        if (result == null) {
            return null;
        }
        T obj = clazz.newInstance();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            HbaseColumn column = field.getAnnotation(HbaseColumn.class);
            if(column == null) {
                continue;
            }
            String family = column.family();
            String qualifier = column.qualifier();
            if (StringUtils.isBlank(family) || StringUtils.isBlank(qualifier)) {
                continue;
            }
            String fieldName = field.getName();
            String value;
            if (ConstantDO.ROW_KEY.equalsIgnoreCase(family)) {
                value = new String(result.getRow());
            } else {
                byte[] bytes = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
                if(bytes == null) {
                    continue;
                }
                value = new String(bytes, StandardCharsets.UTF_8);
            }
            Method method  = new PropertyDescriptor(fieldName, clazz).getWriteMethod();
            method.invoke(obj, convertType(field.getType(), value));
        }
        return obj;
    }


    private static Object convertType(Class<?> type, String value) throws Exception {
        String typeName = type.getTypeName();
        HbaseFieldTypeEnum typeEnum = HbaseFieldTypeEnum.get(typeName);
        if(typeEnum == null) {
            throw new TypeNotSupportException(typeName);
        }
        if(typeEnum.equals(HbaseFieldTypeEnum.DATE)) {
            return new Date(Long.parseLong(value));
        }
        String methodName = typeEnum.getMethodName();
        if("".equals(methodName)) {
            return value;
        }
        if("new".equals(methodName)) {
            return Class.forName(typeEnum.getTypeName()).getConstructor(String.class).newInstance(value);
        }
        return Class.forName(typeEnum.getTypeName()).getMethod(methodName, String.class).invoke(null, value);
    }

    public static List<HbaseCell> resultToHbaseCell(Result result) {
        List<HbaseCell> hbaseCellList = new ArrayList<>();
        if(result == null) {
            return hbaseCellList;
        }
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            HbaseCell hbaseCell = parseCell(cell);
            if(hbaseCell != null) {
                hbaseCellList.add(hbaseCell);
            }
        }
        return hbaseCellList;
    }

    private static HbaseCell parseCell(Cell cell) {
        try {
            HbaseCell hbaseCell = new HbaseCell();
            hbaseCell.setFamily(Bytes.toStringBinary(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            hbaseCell.setQualifier(Bytes.toStringBinary(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            hbaseCell.setTimestamp(cell.getTimestamp());
            hbaseCell.setValue(new String(CellUtil.cloneValue(cell)));
            hbaseCell.setVlen(cell.getValueLength());
            return hbaseCell;
        } catch (Exception e) {
            log.error("parseCell_err:{}", cell ,e);
            return null;
        }
    }

    public static HashMap<String, Map<String, String>> resultToMap(Result result) {
        HashMap<String, Map<String, String>> mapAll = new HashMap<>();
        if(result == null) {
            return mapAll;
        }
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            HbaseCell hbaseCell = parseCell(cell);
            if(hbaseCell == null) {
                continue;
            }
            Map<String, String> map = new HashMap<>();
            map.put(hbaseCell.getQualifier(), hbaseCell.getValue());
            if(!mapAll.containsKey(hbaseCell.getFamily())) {
                mapAll.put(hbaseCell.getFamily(), map);
                continue;
            }
            mapAll.get(hbaseCell.getFamily()).put(hbaseCell.getQualifier(), hbaseCell.getValue());
        }
        return mapAll;
    }

}

